
package cn.tang.tframe.client.zookeeper;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.tang.tframe.common.AbstractLifecycle;
import cn.tang.tframe.common.base.MainConfigLoader;
import cn.tang.tframe.common.utils.lang.StringUtils;

/**
 * zk客户端 <p/> 创建时间: 14-8-6 下午1:52<br/>
 *
 * @author tanghc
 * @since v0.0.1
 */
public class ZKClient extends AbstractLifecycle {

	public static final Logger logger = LoggerFactory.getLogger(ZKClient.class);

	private CuratorFramework zkClient = null;
	private static ConcurrentMap<String, CuratorFramework> zkClientMap =
		new ConcurrentHashMap<String, CuratorFramework>();

	private ZKClient() {
		this.start();
	};

	@Override
	protected void doStart() {

		synchronized (ZKClientHolder.class) {
			if (!isStart) {
				isStart = true;
				String url = buildUrl();
				logger.debug("use zk url {} to create client", url);
				zkClient = createAndStartByUrl(url);
				logger.warn("ZKClient start success! {} ", url);
			}
		}
	}

	private String buildUrl() {

		String zkUrl = MainConfigLoader.getInstance().getZkUrl();
		if (StringUtils.isBlank(zkUrl)) {
			throw new RuntimeException("no zk url");
		}
		return zkUrl;
	}

	/**
	 * 根据ip获取zk client, 可以直接调用该方法，但不建议 请使用 ZKClientManager 调用
	 * 
	 * @param ip
	 * @return
	 */
	public static CuratorFramework create(String url) {

		logger.warn(" start conn zk server {} ", url);

		CuratorFramework newClient = zkClientMap.get(url);
		if (null == newClient) {
			newClient = CuratorFrameworkFactory.newClient(
				url, new ExponentialBackoffRetry(1000, 3));

			zkClientMap.putIfAbsent(url, newClient);
			newClient = zkClientMap.get(url);
		}
		logger.warn("  conn zk server {} success!", url);
		return newClient;
	}

	/**
	 * 根据url获取zk client, 可以直接调用该方法，但不建议 请使用 ZKClientManager 调用
	 */
	public static CuratorFramework createAndStartByUrl(String url) {

		CuratorFramework curatorFramework = create(url);
		if (null == curatorFramework) {
			throw new RuntimeException("not create zk client: " + url);
		}
		curatorFramework.start();
		return curatorFramework;
	}

	@Override
	public void stop() {

		if (zkClient != null) {
			zkClient.close();
		}
		// throw new RuntimeException("un implemented");
	}

	private static class ZKClientHolder {

		private static final ZKClient instance = new ZKClient();
	}

	/**
	 * 获取zk客户端实例（单例）
	 * 
	 * @return
	 */
	public static CuratorFramework getClient() {

		// 初始化client
		// ZKClientHolder.instance.start();
		ZKClient instance = ZKClientHolder.instance;
		return instance.zkClient;
	}

	public static void main(String[] args) {

		getClient();
	}

}
